package com.zg.callback;


import com.zg.CliemtManage.ClientCheck;
import com.zg.CliemtManage.ClientInfo;
import com.zg.service.MqttBrokerMsgSave;
import com.zg.utils.SpringContextUtil;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;


import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.stream.Collectors;



@Slf4j
public class BootMqttMsgBack {

    private static MqttBrokerMsgSave mqttBrokerMsgSave = SpringContextUtil.getBean(MqttBrokerMsgSave.class);

    private static SuccessOrFailAck successOrFailAck = new SuccessOrFailAck();

    //管理设备的上线线，订阅主题等等
    private static ClientInfo clientInfo = new ClientInfo();

    private static ClientCheck clientCheck = new ClientCheck();

    /**
     * 确认连接
     * @param channel
     * @param mqttMessage
     */
    public static void connack(Channel channel, MqttMessage mqttMessage){
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage)mqttMessage;
        MqttFixedHeader mqttFixedHeader = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeader = mqttConnectMessage.variableHeader();

        MqttConnectPayload payload = mqttConnectMessage.payload();
        String username = payload.userName();
        String password = "";
        try {
            password = new String(payload.passwordInBytes(), "UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        //对用户进行校验
        if (!clientCheck.checkClient(username,password)){
            MqttConnAckMessage connAck = successOrFailAck.connFailAck(mqttConnectVariableHeader, mqttFixedHeader, MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
            channel.writeAndFlush(connAck);
            channel.close();
            return;
        }

        MqttConnAckMessage connAck = successOrFailAck.connOkAck(mqttConnectVariableHeader, mqttFixedHeader);
        channel.writeAndFlush(connAck);

        log.info("客户端id：{}，channelId:{}",payload.clientIdentifier(),channel.id());

        //设置上线设备管理
        clientInfo.client_accept(payload.clientIdentifier(), channel);
    }


    /**
     * 根据qos发布确认
     * @param channel
     * @param mqttMessage
     */
    public static void puback(Channel channel,MqttMessage mqttMessage){
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
        MqttQoS qos = (MqttQoS) mqttFixedHeaderInfo.qosLevel();
        byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
        mqttPublishMessage.payload().readBytes(headBytes);
        String data = new String(headBytes);
        System.out.println("publish data--"+data);

        MqttPublishVariableHeader variableHeader = mqttPublishMessage.variableHeader();
        String topicName = variableHeader.topicName();

        if (mqttFixedHeaderInfo.isRetain()){
            mqttBrokerMsgSave.saveMsg(data);
        }

        //转发消息
        clientInfo.relayMessage(topicName,data);

        log.info("发送消息到服务端的channelId:{}",channel.id());
        switch (qos) {
            case AT_MOST_ONCE: 		//	至多一次
                break;
            case AT_LEAST_ONCE:		//	至少一次
                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK,mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                //	构建PUBACK消息体
                MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                log.info("back--"+pubAck.toString());
                channel.writeAndFlush(pubAck);
                break;
            case EXACTLY_ONCE:		//	刚好一次
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC,false, MqttQoS.AT_LEAST_ONCE,false,0x02);
                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2,mqttMessageIdVariableHeaderBack2);
                log.info("back--"+mqttMessageBack.toString());
                channel.writeAndFlush(mqttMessageBack);
                break;
            default:
                break;
        }

    }



    /**
     * 	发布完成 qos2
     * @param channel
     * @param mqttMessage
     */
    public static void pubcomp (Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP,false, MqttQoS.AT_MOST_ONCE,false,0x02);
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack,mqttMessageIdVariableHeaderBack);
        log.info("back--"+mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }


    /**
     * 	订阅确认
     * @param channel
     * @param mqttMessage
     */
    public static void suback(Channel channel, MqttMessage mqttMessage) {
        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());
        //log.info(topics.toString());
        //返回的订阅成功与否
        List<Integer> grantedQoSLevels = new ArrayList<>(topics.size());
        System.out.println("topic:========"+topics);


        clientInfo.addClientTopic(topics,channel);

        for (String topic : topics) {
            if (topic.equals("testtopic/96")){
                //128表失败
                MqttSubAckMessage subAck = successOrFailAck.subAck(Collections.singletonList(128), topics, variableHeaderBack);
                channel.writeAndFlush(subAck);
                return;
            }
        }

        for (int i = 0; i < topics.size(); i++) {
            grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }
        MqttSubAckMessage subAck = successOrFailAck.subAck(grantedQoSLevels, topics, variableHeaderBack);
        log.info("back--"+subAck.toString());
        channel.writeAndFlush(subAck);
    }

    /**
     * 	取消订阅确认
     * @param channel
     * @param mqttMessage
     */
    public static void unsuback(Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文	可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
        //	构建返回报文	取消订阅确认
        MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack,variableHeaderBack);

        MqttUnsubscribeMessage unsubscribeMessage = (MqttUnsubscribeMessage)mqttMessage;
        MqttUnsubscribePayload payload = unsubscribeMessage.payload();
        List<String> topics = payload.topics();
        clientInfo.delClientTopic(channel,topics);

        log.info("back--"+unSubAck.toString());
        channel.writeAndFlush(unSubAck);
    }

    /**
     * 	心跳响应
     * @param channel
     * @param mqttMessage
     */
    public static void pingresp (Channel channel, MqttMessage mqttMessage) {
        //	心跳响应报文	11010000 00000000  固定报文
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
        log.info("back--"+mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }

    public static void disConnect(Channel channel, MqttMessage mqttMessage) {
        log.info("客户端离线消息报:{}",mqttMessage);

        clientInfo.client_shutdown(channel);
    }
}
